package spark.sql

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
import org.apache.spark.sql.{DataFrame, Dataset, Row, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}

/**
 * Created by ibf on 2018/2/4.
 */
case class Person(name:String,age:Int)
object RDD2DataFrame {
  def main(args: Array[String]) {
    //1、创建spark上下文
    val conf = new SparkConf()
      .setAppName("HiveJoinMysqlDataSparkSql")
      .setMaster("local[*]")
    val sc = SparkContext.getOrCreate(conf)
    //这里要考虑是否要读取hive的数据，或者使用HQL
    //如果需要：就使用HiveContext，如果不需要，就使用SQLContext
    val sqlContext = new SQLContext(sc)
    /**
     * 第一：DataFrame转换成RDD
     *    直接通过.rdd进行转换
     */
    val jsonPath = "datas/people.json"
    val jsonDataFrame = sqlContext.jsonFile(jsonPath)
    //通过DataFrame的.rdd API将DataFrame转换为RDD[Row]类型
    val jsonRdd: RDD[Row] = jsonDataFrame.rdd
    val resultRdd = jsonRdd.map(row => (row.getAs[Long](0),row.getAs[String]("name")))
    resultRdd.foreach(println)
    println("==========================")
    /**
     * 第二：rdd转换成DataFrame
     *    方法一：通过反射机制
     *      要求rdd中的数据类型必须是case-class类型
     *      要求引入sqlContext中的隐式转换函数
     */
    import sqlContext.implicits._
    val rdd: RDD[Person] = sc.parallelize(Array(
      Person("xiaoming",32),
      Person("jim",23),
      Person("tom",19),
      Person("link",32),
      Person("leo",20)
    ))
    //通过toDF()或者toDS()就可以将rdd转换成DataFrame或者DataSet
    val df: DataFrame = rdd.toDF()
    val ds: Dataset[Person] = rdd.toDS
    //可以手动给定字段名称，但是字段名称的参数个数，一定要和case class的属性个数相等，否则会报错
    val df2 = rdd.toDF("name1","age1")

    //数据展示
    df.show()
    println("==========================")
    df2.show()
    println("============DS==============")
    ds.show()

    /**
     * 方法二：转换的时候，明确给定字段和schema信息
     */
    val rowRDD: RDD[Row] = sc.parallelize(Array(
      ("xiaoming",32,"man"),
      ("jim",23,"female"),
      ("tom",19,"man"),
      ("link",32,"man"),
      ("leo",20,"female")
    )).map{
      case (name,age,sex) =>
        Row(name,age,sex)
    }
    //自定义schema信息
    /**
     *     name: String,  字段名称
          dataType: DataType, 字段类型，可以进入到DataType中，使用Ctrl+H查看其自雷
          nullable: Boolean = true, 字段是否能为空，可以不给定，默认是能为空的
          metadata: Metadata = Metadata.empty  字段的元数据信息，可以不给定
     */
    val schema = StructType(Array(
      StructField("name3",StringType),
      StructField("age3",IntegerType),
      StructField("sex3",StringType)
    ))

    // def createDataFrame(rowRDD: RDD[Row], schema: StructType)
    val df3 = sqlContext.createDataFrame(rowRDD,schema)
    println("==========================")
    df3.show


  }
}
